package com.sh.data.engine.domain.datadev.flink.service.impl;

import cn.hutool.core.convert.Convert;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.FlinkJobStatusEnum;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkJobDomain;
import com.sh.data.engine.domain.datadev.flink.model.domain.FlinkLogRecordDomain;
import com.sh.data.engine.domain.datadev.flink.service.FlinkJobService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeConfigService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkNodeService;
import com.sh.data.engine.domain.datadev.flink.service.FlinkSqlTaskService;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.repository.dao.datadev.FlinkJobMapper;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkJobRecordEntity;
import com.sh.data.engine.repository.dao.datadev.entity.FlinkMenuNodeEntity;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author：jingchen.hr
 * @descripion:
 * @data:6/17/202211:46 AM
 */
@Service
public class FlinkJobServiceImpl extends ServiceImpl<FlinkJobMapper, FlinkJobEntity>
    implements FlinkJobService {

    @Autowired
    private UserService userService;

    @Autowired
    private FlinkSqlTaskService flinkSqlTaskService;
    @Autowired
    private FlinkNodeConfigService flinkNodeConfigService;
    @Autowired
    private FlinkNodeService flinkNodeService;

    @Override
    public Long addJob(
        Long projectId,
        Long nodeId,
        Long versionId,
        Integer taskType,
        Integer execType,
        Date planTime) {
        FlinkJobEntity flinkJobEntity = this.getJobByNodeIdAndExecType(nodeId, execType);
        if (null == flinkJobEntity) {
            flinkJobEntity = new FlinkJobEntity();
            flinkJobEntity.setProjectId(projectId);
            flinkJobEntity.setNodeId(nodeId);
            flinkJobEntity.setNodeConfigId(Long.valueOf(versionId));
            flinkJobEntity.setExecType(execType);
            flinkJobEntity.setTaskType(taskType);
            flinkJobEntity.setStatus(FlinkJobStatusEnum.RUNNING.getCode());
            flinkJobEntity.setPlanTime(planTime);
            flinkJobEntity.setStartTime(new Date());
            flinkJobEntity.setCreateTime(new Date());
            this.save(flinkJobEntity);
        } else {
            FlinkJobEntity entity = new FlinkJobEntity();
            entity.setId(flinkJobEntity.getId());
            entity.setNodeConfigId(Long.valueOf(versionId));
            entity.setPlanTime(planTime);
            entity.setStartTime(new Date());
            entity.setStatus(FlinkJobStatusEnum.RUNNING.getCode());
            entity.setFlinkJobId(null);
            entity.setEndTime(null);

            UpdateWrapper<FlinkJobEntity> wrapper = Wrappers.update();
            wrapper
                .lambda()
                .eq(FlinkJobEntity::getId, flinkJobEntity.getId())
                .set(FlinkJobEntity::getEndTime, null)
                .set(FlinkJobEntity::getFlinkJobId, null);

            this.update(entity, wrapper);
        }

        return flinkJobEntity.getId();
    }

    private FlinkJobEntity getJobByNodeIdAndExecType(Long nodeId, Integer execType) {

        // query.addCriteria(Criteria.where("nodeId").is(nodeId).and("execType").is(execType));

        LambdaQueryWrapper<FlinkJobEntity> eq =
            new LambdaQueryWrapper<FlinkJobEntity>()
                .eq(FlinkJobEntity::getNodeId, nodeId)
                .eq(FlinkJobEntity::getExecType, execType);
        FlinkJobEntity one = getOne(eq);
        return one;
    }

    @Override
    public void updateJob(
        Long jobId, LinkedHashSet<String> flinkJobIds, LinkedHashSet<String> yarnApplicationIds) {
        FlinkJobEntity entity = new FlinkJobEntity();
        entity.setId(jobId);
        entity.setFlinkJobId(StringUtils.join(flinkJobIds, ","));
        // entity.setYarnApplicationId(StringUtils.join(yarnApplicationIds, ","));
        this.updateById(entity);
    }

    @Override
    public void updateJobStatus(Long jobId, Integer status) {
        FlinkJobEntity entity = new FlinkJobEntity();
        entity.setId(jobId);
        entity.setStatus(status);

        List<Integer> finishedStatusList =
            Lists.newArrayList(
                FlinkJobStatusEnum.SUCCESS.getCode(),
                FlinkJobStatusEnum.FAILURE.getCode(),
                FlinkJobStatusEnum.STOPPED.getCode(),
                FlinkJobStatusEnum.PAUSED.getCode());

        if (finishedStatusList.contains(status)) {
            entity.setEndTime(new Date());
        }

        this.updateById(entity);
    }

    @Override
    public FlinkJobDomain getByFlinkJobId(Long jobId) {
        final FlinkJobEntity entity = this.getById(jobId);

        return ConvertUtil.copyProperties(entity, FlinkJobDomain.class);
    }

    @Override
    public FlinkJobDomain getByNodeId(Long nodeId, Integer execType) {

        LambdaQueryWrapper<FlinkJobEntity> eq =
            new LambdaQueryWrapper<FlinkJobEntity>()
                .eq(FlinkJobEntity::getNodeId, nodeId)
                .eq(FlinkJobEntity::getExecType, execType);
        final FlinkJobEntity entity = this.getOne(eq);
        return ConvertUtil.copyProperties(entity, FlinkJobDomain.class);
    }

    @Override
    public List<FlinkJobDomain> getRunningJobList() {
        LambdaQueryWrapper<FlinkJobEntity> eq =
            new LambdaQueryWrapper<FlinkJobEntity>()
                .eq(FlinkJobEntity::getStatus, FlinkJobStatusEnum.RUNNING.getCode());
        List<FlinkJobEntity> runningJobList = list(eq);
        return ConvertUtil.copyProperties(runningJobList, FlinkJobDomain.class);
    }

    @Override
    @Deprecated
    public List<FlinkJobRecordEntity> getJobStatusList(Long projectId, Integer taskType, Date start) {

        //    Query query = new Query();
        //    Criteria criteria = new Criteria();
        //    criteria.and("execType").is(1).and("startTime").gt(start);
        //    if (Objects.nonNull(taskType)) {
        //      criteria.and("taskType").is(taskType);
        //    }
        //    if (StringUtils.isNotBlank(projectId)) {
        //      criteria.and("projectId").is(projectId);
        //    }
        //
        //    return findMulti(query);

        LambdaQueryWrapper<FlinkJobEntity> gt =
            new LambdaQueryWrapper<FlinkJobEntity>()
                .eq(FlinkJobEntity::getExecType, 1)
                .gt(FlinkJobEntity::getStartTime, start);
        if (Objects.nonNull(taskType)) {
            gt.eq(FlinkJobEntity::getTaskType, taskType);
        }
        if (Objects.nonNull(projectId)) {
            gt.eq(FlinkJobEntity::getProjectId, projectId);
        }
        List<FlinkJobEntity> jobStatusList = this.list(gt);
        Map<Integer, List<FlinkJobEntity>> collect =
            jobStatusList.stream().collect(Collectors.groupingBy(FlinkJobEntity::getStatus));
        return null;
    }

    @Override
    public PageResult<FlinkLogRecordDomain> getTaskRecordPageInfo(
        Long projectId,
        String sort,
        Integer pageNum,
        Integer pageSize,
        String taskName,
        Integer taskStatus,
        Integer taskType) {
        //    Criteria criteria = Criteria.where("projectId").is(projectId);//.and("execType").is(1);
        //
        //    if (Objects.nonNull(taskStatus)) {
        //      criteria.and("status").is(taskStatus);
        //    }
        //    if (Objects.nonNull(taskType)) {
        //      criteria.and("execType").is(taskType);
        //    }
        //    Query query = new Query(criteria);
        //
        //    if (Objects.nonNull(sort)) {
        //      query.with(sort);
        //    }

        LambdaQueryWrapper<FlinkJobEntity> eq =
            new LambdaQueryWrapper<FlinkJobEntity>().eq(FlinkJobEntity::getProjectId, projectId);

        if (Objects.nonNull(taskStatus)) {
            eq.eq(FlinkJobEntity::getStatus, taskStatus);
        }
        if (Objects.nonNull(taskType)) {
            eq.eq(FlinkJobEntity::getExecType, taskType);
        }
        eq.orderByDesc(FlinkJobEntity::getCreateTime);

        IPage<FlinkJobEntity> page = page(new Page<>(pageNum, pageSize), eq);

        List<FlinkJobEntity> result = page.getRecords();

        List<Long> nodeConfigIds =
            result.stream().map(i -> i.getNodeConfigId()).distinct().collect(Collectors.toList());
        List<Long> nodeIds =
            result.stream().map(i -> i.getNodeId()).distinct().collect(Collectors.toList());
        List<FlinkMenuNodeEntity> menuNodeList = Lists.newArrayList();
        if (CollectionUtils.isNotEmpty(nodeIds)) {
            menuNodeList = flinkNodeService.listByIds(nodeIds);
        }
        final List<String> userIds =
            menuNodeList.stream().map(i -> i.getCreatorId()).collect(Collectors.toList());
        Map<String, String> userInfoMap = userService.getMapByUserIds(userIds);
        Map<Long, FlinkMenuNodeEntity> nodeInfoMap =
            menuNodeList.stream().collect(Collectors.toMap(i -> i.getId(), Function.identity()));
        Map<Long, Integer> scheduleTypeMap =
            flinkNodeConfigService.getScheduleTypeMapByIds(nodeConfigIds);

        List<FlinkLogRecordDomain> list = Lists.newArrayList();
        for (FlinkJobEntity i : result) {
            FlinkMenuNodeEntity one = nodeInfoMap.get(i.getNodeId());
            if (Objects.isNull(one)) {
                continue;
            }
            FlinkLogRecordDomain convert = Convert.convert(FlinkLogRecordDomain.class, i);
            if (StringUtils.isNotBlank(taskName) && !one.getName().contains(taskName)) {
                continue;
            }
            String userName = userInfoMap.get(one.getCreatorId());
            convert.setUserName(userName);
            convert.setTaskName(one.getName());
            Date startTime = i.getStartTime();
            // Date endTime = Objects.nonNull(i.getEndTime()) ? new Date() : i.getEndTime();
            Date endTime = i.getEndTime();
            Long expendTime = null;
            if (Objects.nonNull(endTime) && Objects.nonNull(startTime)) {
                expendTime = (endTime.getTime() - startTime.getTime()) / 1000;
            }
            convert.setExpendTime(expendTime);
            convert.setScheduleType(scheduleTypeMap.get(i.getNodeConfigId()));
            convert.setStatus(i.getStatus());
            list.add(convert);
        }

        PageResult<FlinkLogRecordDomain> flinkLogRecordDomainPageResult =
            new PageResult<>(list, pageNum, pageSize);

        return flinkLogRecordDomainPageResult;
    }

    @Override
    public FlinkJobEntity getDetailById(Long id) {

        return this.getById(id);
    }

    @Override
    public List<FlinkJobEntity> getJobsByProjectId(Long projectId) {

        LambdaQueryWrapper<FlinkJobEntity> eq =
            new LambdaQueryWrapper<FlinkJobEntity>().eq(FlinkJobEntity::getProjectId, projectId);

        return this.list(eq);
    }
}
